package cn.smileyan.demos.operators;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;

/**
 * 例子2：flink 的 flatMap
 * @author smileyan
 */
public class FlatMapDemo1 {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<List<Integer>> fixedInput = env.fromElements(
                Arrays.asList(1, 2, 3, 4),
                Arrays.asList(5, 6),
                Arrays.asList(7, 8, 9, 10, 11));

        SingleOutputStreamOperator<Integer> sums = fixedInput.flatMap(new FlatMapFunction<List<Integer>, Integer>() {
            @Override
            public void flatMap(List<Integer> integers, Collector<Integer> collector) throws Exception {
                int sum = 0;
                for (Integer integer : integers) {
                    sum += integer;
                }
                collector.collect(sum);
            }
        });

        sums.print();

        env.execute("Flink FlatMap Example for Sums");
    }

}
